/**
 * 
 */
package com.serviceprovider.business.module.admin.bonus.services.dao.impl;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.dao.EmptyResultDataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementCreator;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.simple.ParameterizedRowMapper;
import org.springframework.jdbc.core.support.JdbcDaoSupport;
import org.springframework.jdbc.support.DatabaseMetaDataCallback;
import org.springframework.jdbc.support.GeneratedKeyHolder;
import org.springframework.jdbc.support.JdbcUtils;
import org.springframework.jdbc.support.MetaDataAccessException;

import com.serviceprovider.business.module.admin.bonus.model.Bonus;
import com.serviceprovider.business.module.admin.bonus.model.BonusCalculationResult;
import com.serviceprovider.business.module.admin.bonus.model.BonusCalculationResultDetail;
import com.serviceprovider.business.module.admin.bonus.model.BonusDetails;
import com.serviceprovider.business.module.admin.bonus.model.BonusPayment;
import com.serviceprovider.business.module.admin.bonus.model.PayBonusForm;
import com.serviceprovider.business.module.admin.bonus.services.dao.interfaces.IBonusService;
import com.serviceprovider.business.module.admin.bonus.utility.BonusPaymentTask;
import com.serviceprovider.business.module.admin.role.model.Role;
import com.serviceprovider.business.module.admin.role.services.dao.interfaces.IRoleService;
import com.serviceprovider.business.module.admin.user.model.UserProfile;
import com.serviceprovider.business.module.admin.user.services.dao.interfaces.IProfileService;
import com.serviceprovider.business.module.admin.user.services.dao.interfaces.IUserService;
import com.serviceprovider.business.module.common.Page;
import com.serviceprovider.business.module.common.PaginationHelper;
import com.serviceprovider.business.module.user.model.impl.User;
import com.serviceprovider.business.utility.AppUtil;
import org.springframework.stereotype.Repository;

/**
 * @author mrduongnv@gmail.com
 * 
 */

@Repository("bonusService")
public class BonusServiceImpl extends JdbcDaoSupport implements IBonusService,
		DatabaseMetaDataCallback {

	/** Logger for this class and subclasses */
	protected final Log logger = LogFactory.getLog(getClass());

    @Autowired
    public BonusServiceImpl(@Qualifier("jdbcTemplate") JdbcTemplate jdbcTemplate) {
        setJdbcTemplate(jdbcTemplate);
    }

    @Autowired
	private IRoleService roleService;

    @Autowired
	private IUserService userService;

    @Autowired
	private IProfileService profileService;

	private Map tableMaps;

    BonusPaymentTask task = new BonusPaymentTask(this);

	public List<Bonus> findAll() {
		String sql = "select id,name,description,active,bonus_type_id,operator,tableName,columnName,"
				+ "method_id,provider_count,customer_count,transaction_count,pay_per_unit,pay_unit from adm_bonus "; //$NON-NLS-1$

		return find(sql);

	}
	
	private static class BonusRowMapper implements ParameterizedRowMapper<Bonus> {
		
		@Override
		public Bonus mapRow(ResultSet rs, int index) throws SQLException {
			Bonus bonus = new Bonus();
			bonus.setId(rs.getInt("id")); //$NON-NLS-1$
			bonus.setName(rs.getString("name")); //$NON-NLS-1$
			bonus.setDescription(rs.getString("description")); //$NON-NLS-1$
			bonus.setActive(rs.getBoolean("active")); //$NON-NLS-1$
			bonus.setTypeId(rs.getInt("bonus_type_id"));
			bonus.setOperator(rs.getString("operator"));
			bonus.setTableName(rs.getString("tableName"));
			bonus.setColumnName(rs.getString("columnName"));
			bonus.setMethodId(rs.getInt("method_id"));
			bonus.setProviderCount(rs.getInt("provider_count"));
			bonus.setCustomerCount(rs.getInt("customer_count"));
			bonus.setTransactionCount(rs.getInt("transaction_count"));
			bonus.setPayPerUnit(rs.getDouble("pay_per_unit"));
			bonus.setPayUnit(rs.getString("pay_unit"));
			return bonus;
		}		
	}
	
	public Page<Bonus> findAll(int pageNo, int recordSize,String sortColumn,String sortOrder) {
		PaginationHelper<Bonus> pageHelper = new PaginationHelper<Bonus>();		
		ParameterizedRowMapper<Bonus> mapper = new BonusRowMapper();
		
		String sql = "select id,name,description,active,bonus_type_id,operator,tableName,columnName,"
				+ "method_id,provider_count,customer_count,transaction_count,pay_per_unit,pay_unit from adm_bonus  order by "+sortColumn+" "+sortOrder; //$NON-NLS-1$
		String countSql="SELECT COUNT(*) FROM adm_bonus";
		return pageHelper.fetchPage(getJdbcTemplate(),
				countSql, sql, new Object[]{}, pageNo, recordSize,sortColumn,sortOrder, mapper);

	}

	@Override
	public Map getTables() {
		try {
			if (this.tableMaps == null)
				this.tableMaps = (Map) JdbcUtils.extractDatabaseMetaData(this
						.getDataSource(), this);
		} catch (Exception ex) {
			throw new RuntimeException("There was error retreiving table list.");
		}
		return this.tableMaps;
	}

	public List<Bonus> find(String sql) {
		List list = getJdbcTemplate().queryForList(sql);

		List<Bonus> ret = new ArrayList<Bonus>();
		Map row = null;
		Bonus aux = null;
		for (Object obj : list) {
			row = (Map) obj;
			aux = new Bonus();
			aux.setId(new Integer(String.valueOf(row.get("id")))); //$NON-NLS-1$
			aux.setName(String.valueOf(row.get("name"))); //$NON-NLS-1$
			aux.setDescription(String.valueOf(row.get("description")));
			aux.setActive(AppUtil.getBoolean(row.get("active")));
			aux.setTypeId(new Integer(String.valueOf(row.get("bonus_type_id"))));
			aux.setOperator(String.valueOf(row.get("operator")));
			aux.setTableName(String.valueOf(row.get("tableName")));
			aux.setColumnName(String.valueOf(row.get("columnName")));

			aux.setMethodId(new Integer(String.valueOf(row.get("method_id"))));
			aux.setProviderCount(AppUtil.getInt(row.get("provider_count")));
			aux.setCustomerCount(AppUtil.getInt(row.get("customer_count")));
			aux.setTransactionCount(AppUtil.getInt(row.get("transaction_count")));
			aux.setPayPerUnit(AppUtil.getDouble(row.get("pay_per_unit")));
			aux.setPayUnit(String.valueOf(row.get("pay_unit")));

			aux.setBonusPayments(this.findBonusPaymentsByBonusId(aux.getId()));
			loadDetails(aux);
			ret.add(aux);
		}
		return ret;

	}

    @SuppressWarnings("rawtypes")
    public List<Bonus> findBonusesByUser(int userId) {
		String sql = "select distinct(b.bonus_id) from adm_bonus_apply b,user u where ((b.role_id=u.role) or (b.user_id=u.id)) and u.id = "
				+ userId;

		List list = getJdbcTemplate().queryForList(sql);
		List<Bonus> ret = new ArrayList<Bonus>();

		List<Integer> bonusIds = new ArrayList<Integer>();
		for (Object obj : list) {
		    Map row = (Map) obj;
		    bonusIds.add(Integer.valueOf((String.valueOf(row.get("bonus_id")))));
		}

		ret.addAll(getBonusByIds(bonusIds));

		return ret;

	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
    private Collection<? extends Bonus> getBonusByIds(List<Integer> ids) {
        getTables();

        RowMapper mapper = new RowMapper() {
            public Bonus mapRow(ResultSet rs, int rowNum) throws SQLException {
                Bonus role = new Bonus();
                BonusServiceImpl.mapRow(role, rs);
                return role;
            }
        };
        final String sql = "select id,name,description,active,bonus_type_id,operator,tableName,columnName,"
                + "method_id,provider_count,customer_count,transaction_count,pay_per_unit,pay_unit from adm_bonus"; //$NON-NLS-1$
        StringBuffer buf = new StringBuffer();
        if (ids.size() > 0) {
            buf.append(" where id in (");
            String concat = "";
            for (Integer id : ids) {
                buf.append(concat).append(id);
                concat = ",";
            }
            buf.append(")");
        }
        try {
            List lstBonus = (List) getJdbcTemplate().query(sql + buf.toString(), mapper);
            for (Object obj : lstBonus) {
                Bonus bonus = (Bonus) obj;
                bonus.setBonusPayments(this.findBonusPaymentsByBonusId(bonus.getId()));
                loadDetails(bonus);
            }
            return lstBonus;
        } catch (EmptyResultDataAccessException e) {
            return null;
        }
    }

    /*
	 * (non-Javadoc)
	 * 
	 * @see springjdbc.example.service.BonusService#getBonus(int)
	 */
	public Bonus getBonusById(int id) {
		getTables();
		RowMapper mapper = new RowMapper() {
			public Bonus mapRow(ResultSet rs, int rowNum) throws SQLException {
				Bonus role = new Bonus();
				BonusServiceImpl.mapRow(role, rs);
				return role;
			}
		};
		final String sql = "select id,name,description,active,bonus_type_id,operator,tableName,columnName,"
				+ "method_id,provider_count,customer_count,transaction_count,pay_per_unit,pay_unit from adm_bonus where id = ?"; //$NON-NLS-1$
		try {
			Bonus ret = (Bonus) getJdbcTemplate().queryForObject(sql,
					new Object[] { id }, mapper);
			System.out.println(ret.getId());
			ret.setBonusPayments(this.findBonusPaymentsByBonusId(ret.getId()));
			loadDetails(ret);
			return ret;
		} catch (EmptyResultDataAccessException e) {
			return null;
		}
	}

    private void loadDetails(Bonus b) {
        String sql = "select desired_amount,flat_rate,percent_rate from adm_bonus_detail where bonus_id = " + b.getId(); //$NON-NLS-1$
        List list = getJdbcTemplate().queryForList(sql);

        BonusDetails d = new BonusDetails();
        for (Object obj : list) {
            Map row = (Map) obj;
            d.getDesiredAmounts().add(new Double(((Number) row.get("desired_amount")).doubleValue()));
            d.getFlatRates().add(new Double(((Number) row.get("flat_rate")).doubleValue()));
            d.getPercentRates().add(new Double(((Number) row.get("percent_rate")).doubleValue()));
        }
        b.setDetails(d);

        sql = "select role_id,user_id from adm_bonus_apply where bonus_id = " + b.getId(); //$NON-NLS-1$
        list = getJdbcTemplate().queryForList(sql);

        List<Role> roles = new ArrayList<Role>();
        List<User> users = new ArrayList<User>();
        for (Object obj : list) {
            Map row = (Map) obj;
            Object roleId = row.get("role_id");
            Object userId = row.get("user_id");

            if (roleId != null)
                roles.add(roleService.getRoleById((((Number) roleId)).intValue()));

            if (userId != null)
                users.add(userService.getById((((Number) userId)).intValue()));

        }
        b.setRoles(roles);
        b.setUsers(users);

    }

	private static void mapRow(Bonus role, ResultSet rs) throws SQLException {
		role.setId(rs.getInt("id")); //$NON-NLS-1$
		role.setName(rs.getString("name")); //$NON-NLS-1$
		role.setDescription(rs.getString("description")); //$NON-NLS-1$
		role.setActive(rs.getBoolean("active")); //$NON-NLS-1$
		role.setTypeId(rs.getInt("bonus_type_id"));
		role.setOperator(rs.getString("operator"));
		role.setTableName(rs.getString("tableName"));
		role.setColumnName(rs.getString("columnName"));
		role.setMethodId(rs.getInt("method_id"));
		role.setProviderCount(rs.getInt("provider_count"));
		role.setCustomerCount(rs.getInt("customer_count"));
		role.setTransactionCount(rs.getInt("transaction_count"));
		role.setPayPerUnit(rs.getDouble("pay_per_unit"));
		role.setPayUnit(rs.getString("pay_unit"));
	}

	private void deleteBonusDetails(int bonusId) {
		String sql = "delete from adm_bonus_apply where bonus_id = ?";
		String sql2 = "delete from adm_bonus_detail where bonus_id = ?";

		getJdbcTemplate().update(sql, new Object[] { bonusId });
		getJdbcTemplate().update(sql2, new Object[] { bonusId });
	}

    private void insertBonusDetails(int bonusId, double[] desiredAmounts, double[] flatRates, double[] percentRates) {
        String sql = "insert into adm_bonus_detail (bonus_id,desired_amount,flat_rate,percent_rate) values (?,?,?,?)";

        for (int i = 0; i < desiredAmounts.length; i++) {

            getJdbcTemplate().update(sql, new Object[] { bonusId, desiredAmounts[i], flatRates[i], percentRates[i] });
        }
    }

    private void insertRoles(int bonusId, int[] roleIds, int[] userIds) {
        String sql = "insert into adm_bonus_apply (bonus_id,role_id) values (?,?)";

        for (int roleId : roleIds) {

            getJdbcTemplate().update(sql, new Object[]{bonusId, roleId});
        }

        String sql2 = "insert into adm_bonus_apply (bonus_id,user_id) values (?,?)";

        for (int userId : userIds) {

            getJdbcTemplate().update(sql2, new Object[]{bonusId, userId});
        }
    }

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * springjdbc.example.service.BonusService#insertBonus(springjdbc.example.
	 * domain.Bonus)
	 */
    public void insertBonus(Bonus bonus, double[] desiredAmounts, double[] flatRates, double[] percentRates, int[] roleIds, int[] userIds) {

        GeneratedKeyHolder k = new GeneratedKeyHolder();
        this.getJdbcTemplate().update(new BonusPreparedStatementCreator(bonus), k);

        bonus.setId(new Integer(k.getKey().intValue()));

        insertBonusDetails(bonus.getId(), desiredAmounts, flatRates, percentRates);
        insertRoles(bonus.getId(), roleIds, userIds);
    }

	public void updateBonus(Bonus bonus, double[] desiredAmounts,
			double[] flatRates, double[] percentRates, int[] roleIds,
			int[] userIds) {

		
		deleteBonusDetails(bonus.getId());

		final String sql = "update adm_bonus set name = ?,description = ?,bonus_type_id = ?,operator=?, active = ?, tableName = ? ,columnName = ?,"
				+ "method_id=?,provider_count=?,customer_count=?,transaction_count=?,pay_per_unit=?,pay_unit=? where id = ?"; //$NON-NLS-1$

		getJdbcTemplate().update(
				sql,
				new Object[] { bonus.getName(), bonus.getDescription(),
						bonus.getTypeId(), bonus.getOperator(),
						bonus.isActive(), bonus.getTableName(),
						bonus.getColumnName(), bonus.getMethodId(),
						bonus.getProviderCount(), bonus.getCustomerCount(),
						bonus.getTransactionCount(), bonus.getPayPerUnit(),
						bonus.getPayUnit(), bonus.getId() });

		insertBonusDetails(bonus.getId(), desiredAmounts, flatRates,
				percentRates);
		insertRoles(bonus.getId(), roleIds, userIds);
	}

	public void deleteBonusesById(int[] idList) {
		final String sql = "delete from adm_bonus where id = ?"; //$NON-NLS-1$

		for (int i = 0; i < idList.length; i++) {
			deleteBonusDetails(idList[i]);
			getJdbcTemplate().update(sql,
					new Object[] { new Integer(idList[i]) });
		}
	}

	public BonusCalculationResult computeBonus(Bonus b, int user_id) {
		if (b.getMethodId().intValue() == 1)
			return computeBonusMethod1(b, user_id);
		else
			return computeBonusMethod2(b, user_id);

	}

	public BonusCalculationResult computeBonusMethod1(Bonus b, int user_id) {
		BonusCalculationResult ret = new BonusCalculationResult();
		String sql = "select " + b.getTableName() + "." + b.getColumnName()
				+ " from " + b.getTableName() + " where user_id = " + user_id;

		Double actualAmount = new Double(-1);

		List l = getJdbcTemplate().queryForList(sql);
		Object val = ((Map) l.get(0)).get(b.getColumnName());
		actualAmount = new Double(val.toString());

		List<Double> desiredAmouts = b.getDetails().getDesiredAmounts();
		int selectedIndex = -1;
		Double selectedDesiredAmouts = new Double(-1);

		for (int i = 0; i < desiredAmouts.size(); i++) {
			Double desiredAmount = desiredAmouts.get(i);
			boolean condOK = this.isTrue(actualAmount, b.getOperator(),
					desiredAmount);
			if (condOK && desiredAmount >= selectedDesiredAmouts) {
				selectedIndex = i;
				selectedDesiredAmouts = desiredAmount;
			}
		}

		// we could determined desiredAmouts , now we're picking the right rate.
		if (selectedIndex > -1) {
			if ((b.getTypeId().intValue() == 1)) {
				ret.setCalculatedAmount(b.getDetails().getFlatRates().get(
						selectedIndex));
			}

			if ((b.getTypeId().intValue() == 2)) {
				double percent = b.getDetails().getPercentRates().get(
						selectedIndex).doubleValue();
				ret.setCalculatedAmount(actualAmount * (percent / 100));
			}

			Double a1 = ret.getCalculatedAmount().doubleValue();
			Double a2 = b.getPaidBalance(user_id).doubleValue();
			ret.setUnpaidBalance(new Double(a1 - a2));
		}

		b.setBonusPayments(findBonusPaymentsByBonusId(b.getId(), user_id));

		ret.setActualAmount(actualAmount);
		return ret;
	}

	private boolean isTrue(Double actualAmount, String operator,
			Double desiredAmount) {
		boolean ret = false;

		if (operator.equals("<")) {
			return (actualAmount.doubleValue() < desiredAmount.doubleValue());
		}

		if (operator.equals(">")) {
			return (actualAmount.doubleValue() > desiredAmount.doubleValue());
		}

		return ret;
	}

    public BonusCalculationResult computeBonusMethod2(Bonus b, int user_id) {

        String sql = "select u.id from adm_users u, role r where (u.role = r.id) and referer_id = " + user_id
                + " and (r.name = 'ROLE_SERVICE_PROVIDER')";
        List spList = getJdbcTemplate().queryForList(sql);

        Map<Integer, BonusCalculationResultDetail> detailsMap = new HashMap();

        for (Object obj : spList) {
            Map row = (Map) obj;
            Integer spId = new Integer(row.get("id").toString());
            User user = userService.getById(spId);

            BonusCalculationResultDetail dtl = new BonusCalculationResultDetail();
            dtl.setServiceProviderId(user.getId());
            try {
                UserProfile profile = profileService.getByUser(user.getId());
                dtl.setServiceProviderName(profile.getFirstName() + " " + profile.getLastName());
            } catch (Exception ex) {
                dtl.setServiceProviderName(user.getEmail());
            }

            dtl.setAffiliateId(user_id);
            detailsMap.put(user.getId(), dtl);
        }

        String sql2 = "select id,service_provider,transaction_count from v_customers where affiliate = " + user_id
                + " order by service_provider,email";
        List list = getJdbcTemplate().queryForList(sql2);

        for (Object obj : list) {
            Map row = (Map) obj;
            Integer spId = new Integer(String.valueOf(row.get("service_provider")));

            // if(!detailsMap.containsKey(spId))
            // detailsMap.put(spId, new BonusCalculationResultDetail());

            BonusCalculationResultDetail dtl = detailsMap.get(spId);

            dtl.setCustomerCount(dtl.getCustomerCount() + 1);

            Integer transactionCount = new Integer(String.valueOf(row.get("transaction_count")));
            dtl.setTransactionCount(dtl.getTransactionCount() + transactionCount);
        }

        int serviceProviderCount = detailsMap.size();
        double totalBonusEarned = 0;
        List<BonusCalculationResultDetail> details = new ArrayList<BonusCalculationResultDetail>(detailsMap.values());
        if (serviceProviderCount >= b.getProviderCount()) {

            for (BonusCalculationResultDetail dtl : details) {
                boolean b1 = dtl.getCustomerCount().intValue() >= b.getCustomerCount();
                boolean b2 = dtl.getTransactionCount().intValue() >= b.getTransactionCount();

                if (b1 && b2) {
                    double bonusEarned = 0;
                    if ("Provider".equals(b.getPayUnit())) {
                        bonusEarned = b.getPayPerUnit().doubleValue() * 1;
                    }

                    if ("Customer".equals(b.getPayUnit())) {
                        bonusEarned = b.getPayPerUnit().doubleValue() * dtl.getCustomerCount().intValue();
                    }

                    if ("Transaction".equals(b.getPayUnit())) {
                        bonusEarned = b.getPayPerUnit().doubleValue() * dtl.getTransactionCount().intValue();
                    }

                    dtl.setBonusEarned(bonusEarned);
                    totalBonusEarned += bonusEarned;
                }
            }

        }

        BonusCalculationResult ret = new BonusCalculationResult();
        ret.setDetails(details);

        Double a1 = new Double(totalBonusEarned);
        Double a2 = b.getPaidBalance(user_id).doubleValue();
        ret.setUnpaidBalance(new Double(a1 - a2));

        b.setBonusPayments(findBonusPaymentsByBonusId(b.getId(), user_id));

        return ret;

    }

	public List<BonusPayment> findBonusPaymentsByBonusId(int bonusId) {
		return findBonusPaymentsByBonusId(bonusId, -1);
	}

	public List<BonusPayment> findBonusPaymentsByBonusId(int bonusId,
			int user_id) {
		String sql = "select id,bonus_id,user_id,amount,pay_as,pay_date from adm_bonus_payment where bonus_id = "
				+ bonusId;
		if (user_id != -1) {
			sql += " and user_id = " + user_id;
		}

		List list = getJdbcTemplate().queryForList(sql);

		List<BonusPayment> ret = new ArrayList<BonusPayment>();
		Map row = null;
		BonusPayment aux = null;
		for (Object obj : list) {
			row = (Map) obj;
			aux = new BonusPayment();
			aux.setId(new Integer(String.valueOf(row.get("id")))); //$NON-NLS-1$
			aux.setBonusId(new Integer(String.valueOf(row.get("bonus_id")))); //$NON-NLS-1$
			aux.setUserId(new Integer(String.valueOf(row.get("user_id")))); //$NON-NLS-1$
			aux.setAmount(AppUtil.getDouble(row.get("amount")));
			aux.setPayAs(new Integer(String.valueOf(row.get("pay_as")))); //$NON-NLS-1$
			aux.setPayDate((Date) row.get("pay_date"));

			ret.add(aux);
		}
		return ret;
	}

	public BonusPayment getBonusPayment(int bonusPaymentId) {
		BonusPayment ret = new BonusPayment();
		String sql = "select id,bonus_id,user_id,amount,pay_as,pay_date from adm_bonus_payment where id = "
				+ bonusPaymentId;
		List list = getJdbcTemplate().queryForList(sql);

		Map row = null;
		for (Object obj : list) {
			row = (Map) obj;
			ret = new BonusPayment();
			ret.setId(new Integer(row.get("id").toString())); //$NON-NLS-1$
			ret.setBonusId(new Integer(row.get("bonus_id").toString())); //$NON-NLS-1$
			ret.setUserId(new Integer(row.get("user_id").toString())); //$NON-NLS-1$
			ret.setAmount(AppUtil.getDouble(row.get("amount")));
			ret.setPayAs(new Integer(row.get("pay_as").toString())); //$NON-NLS-1$
			ret.setPayDate((Date) row.get("pay_date"));
			break;
		}
		return ret;
	}

	public void pay(int bonusId, int userId, double payAmount, int payAs) {
		pay(getBonusById(bonusId), userService.getById(userId), payAmount,
				payAs);

	}

	private void pay(Bonus b, User u, Double payAmount, Integer payAs) {
		BonusCalculationResult result = computeBonus(b, u.getId());
		Double amount = result.getUnpaidBalance();

		if (payAmount.doubleValue() > amount.doubleValue())
			throw new RuntimeException("PayAmount greater than UnpaidBalance");

		GeneratedKeyHolder k = new GeneratedKeyHolder();

		BonusPayment p = new BonusPayment();
		p.setBonusId(b.getId());
		p.setUserId(u.getId());
		p.setAmount(payAmount);
		p.setPayAs(payAs);

		getJdbcTemplate().update(new BonusPaymentCreator(p), k);
		Integer paymetId = new Integer(k.getKey().intValue());

		String sql = "insert into adm_bonus_received  (payment_id,cash,credit) values (?,?,?)";
		Double cash = (p.getPayAs().intValue() == 1) ? p.getAmount() : null;
		Double credit = (p.getPayAs().intValue() == 2) ? p.getAmount() : null;

		getJdbcTemplate().update(sql, new Object[] { paymetId, cash, credit });

	}

	public List<User> getUsersByBonusId(int bonusId) {
		String sql = "select role_id,user_id  from adm_bonus_apply where bonus_id = "
				+ bonusId;
		List list = getJdbcTemplate().queryForList(sql);

		Map<Integer, User> ret = new HashMap();

		for (Object obj : list) {
			Map row = (Map) obj;

			Object role_id = row.get("role_id");
			Object user_id = row.get("user_id");

			Role role=new Role();
			role.setId(new Integer(role_id.toString()));
			if (role_id != null) {
				Page<User> users = userService.findByRole(role);
				for (User u : users.getPageItems()) {
					if (!ret.containsKey(u.getId()))
						ret.put(u.getId(), u);
				}
			}

			if (user_id != null) {
				Integer tmpId = new Integer(user_id.toString());
				if (!ret.containsKey(tmpId)) {
					ret.put(tmpId, userService.getById(tmpId));
				}
			}
		}
		return new ArrayList<User>(ret.values());

	}

	public void createPaymentSchedule(PayBonusForm f) {
		String sql = "insert into adm_bonus_schedule (bonus_id,user_id,amount,pay_on,pay_when,pay_as) values (?,?,?,?,?,?)";
		getJdbcTemplate().update(
				sql,
				new Object[] { f.getBonusId(), f.getUserId(), f.getPayAmount(),
						f.getPayOn(), f.getPayWhen(), f.getPayAs() });
	}

	public void executeScheduledPayments() {
		String sql = "select id,bonus_id,user_id,amount,pay_on,pay_when,pay_as  from adm_bonus_schedule";
		List list = getJdbcTemplate().queryForList(sql);

		// java.awt.Toolkit.getDefaultToolkit().beep();

		for (Object obj : list) {
			Map row = (Map) obj;
			Integer id = new Integer(row.get("id").toString());
			Integer bonus_id = new Integer(row.get("bonus_id").toString());
			Integer user_id = new Integer(row.get("user_id").toString());
			Integer pay_as = new Integer(row.get("pay_as").toString());

			Object amount = row.get("amount");
			Object pay_on = row.get("pay_on");
			Object pay_when = row.get("pay_when");

			if (pay_on != null) {
				Date d = (Date) pay_on;
				Date today = new Date();
				if (!today.before(d)) {
					pay(bonus_id, user_id, new Double(amount.toString()),
							pay_as);
					deletePaymentSchedules(id);
				}

			}

			if (pay_when != null) {
				Bonus b = getBonusById(bonus_id);
				BonusCalculationResult result = computeBonus(b, user_id);
				double testVal = new Double(pay_when.toString()).doubleValue();
				if (result.getUnpaidBalance().doubleValue() >= testVal) {
					pay(bonus_id, user_id, result.getUnpaidBalance(), pay_as);
					deletePaymentSchedules(id);

				}

			}

		}
	}

	private void deletePaymentSchedules(int id) {
		String sql = "delete from adm_bonus_schedule where id = ?";
		getJdbcTemplate().update(sql, new Object[] { id });

	}

	private boolean isNumericDataType(int sqlDataType) {
		boolean ret = false;

		ret = (sqlDataType == Types.BIGINT || sqlDataType == Types.DECIMAL
				|| sqlDataType == Types.DOUBLE || sqlDataType == Types.FLOAT
				|| sqlDataType == Types.INTEGER || sqlDataType == Types.NUMERIC
				|| sqlDataType == Types.DECIMAL || sqlDataType == Types.REAL
				|| sqlDataType == Types.SMALLINT || sqlDataType == Types.TINYINT);
		return ret;
	}

	@Override
	public Object processMetaData(DatabaseMetaData dbmd) throws SQLException,
			MetaDataAccessException {
		String[] types = { "TABLE" };
		ResultSet rset = dbmd.getTables(null, null, null, types);
		Map tableMap = new HashMap();
		while (rset.next()) {
			String tableName = rset.getString("TABLE_NAME");
			ResultSet cRset = dbmd.getColumns(null, null, tableName, null);
			List columns = new ArrayList();
			while (cRset.next()) {
				String columnName = cRset.getString("COLUMN_NAME");
				int dataType = cRset.getInt("DATA_TYPE");
				if (isNumericDataType(dataType))
					columns.add(cRset.getString("COLUMN_NAME"));
			}
			tableMap.put(tableName, columns);
		}
		return tableMap;
	}

}

class BonusPreparedStatementCreator implements PreparedStatementCreator {
	private Bonus bonus;

	public BonusPreparedStatementCreator(Bonus b) {
		bonus = b;
	}

	public PreparedStatement createPreparedStatement(Connection conn)
			throws SQLException {
		String sql = "insert into adm_bonus (name,description,bonus_type_id,operator,active,tableName,columnName,"
				+ "method_id,provider_count,customer_count,transaction_count,pay_per_unit,pay_unit) values(?,?,?,?,?,?,?,?,?,?,?,?,?)";
		PreparedStatement ret = conn.prepareStatement(sql,
				Statement.RETURN_GENERATED_KEYS);
		ret.setString(1, bonus.getName());
		ret.setString(2, bonus.getDescription());
		ret.setInt(3, bonus.getTypeId());
		ret.setString(4, bonus.getOperator());
		ret.setBoolean(5, bonus.isActive());
		ret.setString(6, bonus.getTableName());
		ret.setString(7, bonus.getColumnName());

		ret.setInt(8, bonus.getMethodId());

		if (bonus.getProviderCount() == null)
			ret.setNull(9, Types.INTEGER);
		else
			ret.setInt(9, bonus.getProviderCount());

		if (bonus.getCustomerCount() == null)
			ret.setNull(10, Types.INTEGER);
		else
			ret.setInt(10, bonus.getCustomerCount());

		if (bonus.getTransactionCount() == null)
			ret.setNull(11, Types.INTEGER);
		else
			ret.setInt(11, bonus.getTransactionCount());

		if (bonus.getPayPerUnit() == null)
			ret.setNull(12, Types.DOUBLE);
		else
			ret.setDouble(12, bonus.getPayPerUnit());

		if (bonus.getPayUnit() == null)
			ret.setNull(13, Types.VARCHAR);
		else
			ret.setString(13, bonus.getPayUnit());

		return ret;
	}

}

class BonusPaymentCreator implements PreparedStatementCreator {
	private BonusPayment payment;

	public BonusPaymentCreator(BonusPayment bp) {
		payment = bp;
	}

	public PreparedStatement createPreparedStatement(Connection conn)
			throws SQLException {
		String sql = "insert into adm_bonus_payment (bonus_id,user_id,amount,pay_as,pay_date) values(?,?,?,?,?)";
		PreparedStatement ret = conn.prepareStatement(sql,
				Statement.RETURN_GENERATED_KEYS);

		ret.setInt(1, payment.getBonusId());
		ret.setInt(2, payment.getUserId());
		ret.setDouble(3, payment.getAmount());
		ret.setInt(4, payment.getPayAs());
		ret.setDate(5, new java.sql.Date(System.currentTimeMillis()));

		return ret;
	}

}
